package org.lib.kafkahello.try1;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import lombok.extern.slf4j.Slf4j;

/**
 * kafka试验1的配置
 * @author ben
 * @date 2021-08-24 16:33:55 CST
 */
@Configuration
@Slf4j
public class Try1Config {

	/**
	 * 主题01：topic01
	 */
	public static final String TOPIC_01 = "topic01";
	/**
	 * 主题01-KEY：topic01_key
	 */
	public static final String TOPIC_01_KEY = "topic01_key";
	/**
	 * 消费者组：topic01g01
	 */
	public static final String TOPIC_01_G01 = "topic01g01";
	
	/**
	 * 监听主题 topic01 的消息
	 * @author ben
	 * @date 2021-08-24 16:33:39 CST
	 * @param record
	 */
	@KafkaListener(id="listenerA",topics = {TOPIC_01}, groupId = TOPIC_01_G01)
	public void listener01(ConsumerRecord<?, ?> record, Acknowledgment ack) {
		log.info("try1-消费-A：topic={}, partition={}, offset={}, key=[{}], value=[{}]",
				record.topic(), record.partition(), record.offset(), record.key(), record.value());
		
		// 消息确认
		// 配置手动确认后，若不执行下面的语句，启动后还会消费
		ack.acknowledge();
	}

//	/**
//	 * 第二个监听器
//	 * @author ben
//	 * @date 2021-08-24 16:59:21 CST
//	 * @param record
//	 */
//	@KafkaListener(id="listenerB", topics = {TOPIC_01}, groupId = TOPIC_01_G01)
//	public void listener02(ConsumerRecord<?, ?> record) {
//		log.info("try1-消费-B：topic={}, partition={}, offset={}, key=[{}], value=[{}]",
//				record.topic(), record.partition(), record.offset(), record.key(), record.value());
//	}
	
}
